package com.yeeiee.source.impl;

import com.yeeiee.config.MysqlSourceConfig;
import com.yeeiee.source.AbstractJdbcSource;
import com.yeeiee.source.ClinkSource;
import lombok.NoArgsConstructor;
import lombok.val;
import org.apache.flink.configuration.Configuration;

@NoArgsConstructor
public class MysqlSource extends AbstractJdbcSource {

    private String catalogName;
    private String databaseName;
    private String tableName;

    public MysqlSource(String catalogName, String tablePath) {
        this.catalogName = catalogName;
        if (!tablePath.contains(".")) {
            throw new IllegalArgumentException("tablePath type must be database.tableName");
        }

        val splitArr = tablePath.split("\\.");
        databaseName = splitArr[0];
        tableName = splitArr[1];

    }

    @Override
    public String factoryIdentifier() {
        return "mysqlSource";
    }

    @Override
    public ClinkSource build(Configuration configuration) {
        super.validate(configuration);

        val catalogName = configuration.get(MysqlSourceConfig.catalog);
        val tablePath = configuration.get(MysqlSourceConfig.tablePath);


        return new MysqlSource(catalogName, tablePath);
    }

    @Override
    public String getTableFullPath() {
        return this.quoteIdentifier(this.catalogName) + "." + this.quoteIdentifier(this.databaseName, this.tableName);
    }

    @Override
    public String quoteIdentifier(String tableName) {
        return "`" + tableName + "`";
    }
}
